package cn.doitedu.demo9;


import cn.doitedu.pojo.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/21
 * @Desc: 学大数据，上多易教育
 *
 *   实现了：可动态注入新的模型和新模型的规则
 **/

@Slf4j
public class Demo9 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置一些必选参数
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);


        // 构建一个kafka source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setTopics("user-events-log")
                .setBootstrapServers("doitedu:9092")
                .setGroupId("g001")
                //.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 将source添加到env，得到数据流
        DataStreamSource<String> jsonStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s1");

        // json解析，封装成javaBean
        SingleOutputStreamOperator<UserEvent> beanStream = jsonStream.map(value -> JSON.parseObject(value, UserEvent.class));

        // 将数据keyBy(用户)
        KeyedStream<UserEvent, Long> keyByStream = beanStream.keyBy(UserEvent::getUid);


        /* *
         * 用 mysql-cdc连接器（source），去抓取 规则元数据信息表中的一切变化操作（增、删、改）
         */
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("doit41") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("doit41.rule_meta") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();


        DataStreamSource<String> ruleMetaCdcStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "rule_meta_source");

        SingleOutputStreamOperator<RuleMetaBean> ruleMetaBeanStream = ruleMetaCdcStream.map(new MapFunction<String, RuleMetaBean>() {
            @Override
            public RuleMetaBean map(String cdcJson) throws Exception {

                RuleMetaBean ruleMetaBean = new RuleMetaBean();

                // 提取op
                JSONObject jsonObject = JSON.parseObject(cdcJson);
                String op = jsonObject.getString("op");
                ruleMetaBean.setOp(op);

                // 提取表数据
                JSONObject dataObject;
                if (op.equals("d")) {
                    dataObject = jsonObject.getJSONObject("before");
                } else {
                    dataObject = jsonObject.getJSONObject("after");
                }

                // 提取规则id
                /**
                 *     "rule_id": "rule-1-002",
                 *     "model_id": "model-1",
                 *     "rule_param_json": "{bbbbbbbbbbbbb}",
                 *     "rule_status": 0
                 */
                String ruleId = dataObject.getString("rule_id");
                String modelId = dataObject.getString("model_id");
                String ruleParamJson = dataObject.getString("rule_param_json");
                int ruleStatus = dataObject.getIntValue("rule_status");

                // 提取规则的预圈选人群bitmap的序列化字节
                byte[] preSelectUsersBitmapBytes = dataObject.getBytes("pre_select_users");
                Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf();
                if(preSelectUsersBitmapBytes !=null) {
                    bitmap.deserialize(ByteBuffer.wrap(preSelectUsersBitmapBytes));
                }

                // 提取规则的运算机代码
                String calculatorCode = dataObject.getString("calculator_code");


                // 把提取到的信息，填充到ruleMetaBean对象
                ruleMetaBean.setRule_id(ruleId);
                ruleMetaBean.setModel_id(modelId);
                ruleMetaBean.setRule_param_json(ruleParamJson);
                ruleMetaBean.setRule_status(ruleStatus);
                ruleMetaBean.setPreSelectUsersBitmap(bitmap);
                ruleMetaBean.setCalculatorCode(calculatorCode);


                return ruleMetaBean;
            }
        });

        // 广播规则元数据流
        BroadcastStream<RuleMetaBean> metaBeanBroadcastStream = ruleMetaBeanStream.broadcast(new MapStateDescriptor<String, String>("bc_state", String.class, String.class));

        // 连接 行为事件keyBy主流  和   规则元数据广播流
        SingleOutputStreamOperator<String> messageStream = keyByStream.connect(metaBeanBroadcastStream)
                .process(new KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, String>() {

                    HashMap<String, RuleCalculator> calculatorPool = new HashMap<>();
                    GroovyClassLoader groovyClassLoader;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        groovyClassLoader = new GroovyClassLoader();
                    }

                    @Override
                    public void processElement(UserEvent userEvent, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {

                        Collection<RuleCalculator> calculators = calculatorPool.values();
                        for (RuleCalculator calculator : calculators) {
                            calculator.calc(userEvent, out);
                        }

                    }

                    @Override
                    public void processBroadcastElement(RuleMetaBean ruleMetaBean, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, String>.Context ctx, Collector<String> out) throws Exception {

                        // 增加规则
                        if (
                                (
                                        ruleMetaBean.getOp().equals("c")
                                                || ruleMetaBean.getOp().equals("r")
                                                || ruleMetaBean.getOp().equals("u")
                                )
                                        && ruleMetaBean.getRule_status() == 1
                        ) {

                            // 根据规则运算机源代码，构建运算机对象
                            Class aClass = groovyClassLoader.parseClass(ruleMetaBean.getCalculatorCode());
                            RuleCalculator calculator = (RuleCalculator) aClass.newInstance();


                            // 初始化运算机对象
                            calculator.init(getRuntimeContext(), ruleMetaBean.getRule_param_json(),ruleMetaBean.getPreSelectUsersBitmap());

                            // 放入运算机池
                            calculatorPool.put(ruleMetaBean.getRule_id(), calculator);

                            log.warn("增加了一个规则运算机，规则id为：{}", ruleMetaBean.getRule_id());
                        }

                        // 删除、下线规则
                        if (ruleMetaBean.getOp().equals("d") || (ruleMetaBean.getOp().equals("u") && ruleMetaBean.getRule_status() == 0)) {
                            calculatorPool.remove(ruleMetaBean.getRule_id());

                            log.warn("下线了一个规则运算机，规则id为：{}", ruleMetaBean.getRule_id());
                        }
                    }
                });


        messageStream.print();

        env.execute();

    }

}
